package com.spark.analysis

import com.alibaba.fastjson.JSON
import org.apache.spark.{SparkConf, SparkContext}

//case class OrderInfo(orderId: String,productId:Int,productPrice:Int)

object TestInsertRedis {

  //每件商品总销售额
  val orderTotalKey = "bussiness::order::total"
  //总销售额
  val totalKey = "bussiness::order::all"
  //Redis数据库
  val dbIndex = 0

  def main(args: Array[String]): Unit = {

    val sparkConf = new SparkConf().setMaster("local[2]").setAppName("TestInsertRedis")
    val sc = new SparkContext(sparkConf)

    val dataRdd = sc.textFile("file:///D:\\JavaProjects\\RealtimeSparkCase\\sparkanalysis\\data\\data.txt")
    dataRdd
      .map(line => {
        val jSONObject = JSON.parseObject(line)
        (jSONObject.getString("productId"), (1, jSONObject.getString("productPrice").toLong))
      })
      .reduceByKey {
        case ((count1, price1), (count2, price2)) => (count1 + count2, price1 + price2) // 按 productId 聚合（累加计数和总价）
      }
      .map {
        case (productId, (count, totalPrice)) => (productId, count, totalPrice) // 转换为 (productId, count, totalPrice)
      }
      .foreach(part => {
        val jedis = RedisClient.pool.getResource() // 获取连接
        try {
          jedis.select(dbIndex)
          // 打印调试日志（确认参数非空）
          println(s"写入 Redis - productId: ${part._1}, count: ${part._2}, totalPrice: ${part._3}")

          // 调用hincrBy方法插入数据后，redis-cli中使用如下方式获取对应key的所有的数据：
          // 127.0.0.1:6379> HGETALL testproduct1
          //1) "product1"   # 字段名（对应 part._1）
          //2) "100"        # 字段值（对应 part._2 的累加结果）
          jedis.hincrBy("test"+part._1, part._1, part._2)
          //
          jedis.incrBy("totalKey", part._3)
        } catch {
          case e: Exception =>
            println(s"Redis 写入失败: ${e.getMessage}")
        } finally {
          jedis.close() // 释放连接回池
        }
      })

    sc.stop()

  }
}
